package com.china.produce.config.kafka;

import com.china.produce.pojo.order.OrderDTO;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;


/**
 * @ClassName
 * @Author niujie
 * @Version
 * @Description
 * @CreateTime 2023/12/2
 */
@Data
@Slf4j
public class KafkaConsumerPool {

    /**
     * 线程并发数
     */
    private Integer concurrentSize = 3;

    /**
     * 工作线程
     */
    private List<Thread> workThreads;


    private List<ConcurrentLinkedQueue> queues;


    private AtomicLong pendingOffsets;


    private volatile boolean stop;


    /**
     * kafka线程名前缀
     */
    private  static final String KAFKA_CONSUMER_WORK_THREAD_PREFIX = "kafka-sort-consumer-thread-";


    public KafkaConsumerPool(KafkaSortConsumerConfig<OrderDTO> config)
    {

        this.workThreads = new ArrayList<>();
        this.concurrentSize = config.getConcurrentSize();
        this.initQueue();
        this.initWorkThread(config.getBizName(), config.getBizService());
        stop = false;
        pendingOffsets = new AtomicLong(0l);
    }

    /**
     * 初始化队列
     */
    public void initQueue()
    {
        this.queues = new ArrayList<>();
        for (int i = 0; i < concurrentSize; i++) {
            this.queues.add(new ConcurrentLinkedQueue());
        }
    }

    /**
     * 初始化线程，有多少个队列就有多个线程处理，一个线程处理一个队列
     *
     * @param bizName
     * @param bizService
     */
    private void initWorkThread(String bizName, Consumer bizService)
    {
        for (int i = 0; i < concurrentSize; i++) {
            String theadName = KAFKA_CONSUMER_WORK_THREAD_PREFIX + bizName + i;
            int num = i;
            Thread thread = new Thread(() -> {
                while (!queues.get(num).isEmpty() || !stop) {
                    OrderDTO order = pollTask(theadName, bizName);
                    if (!Objects.isNull(order)) {
                        CompletableFuture<Boolean> future = CompletableFuture.supplyAsync(() -> {
                            //处理耗时任务
                            //bizService.accept(order);
                            return true;
                        });
                        try {
                            //100ms直接返回，熔断处理
                            future.get(100, TimeUnit.MILLISECONDS);
                        }catch (Exception exception) {

                        }finally {
                            //执行完任务加一，
                            pendingOffsets.incrementAndGet();
                        }
                    }
                }
            }, theadName);
            workThreads.add(thread);
            thread.start();
        }
    }

    /**
     * 提交任务到queues中, queues分类数等于线程数，
     * 就是说每个queue有单独的线程在消费，
     * 这样就保证每个queue中的处理是单线程有顺序的
     *
     * @param id
     * @param task
     */
    public void submitTask(Long id,  OrderDTO task)
    {

        ConcurrentLinkedQueue linkedQueue = queues.get((int) (id % this.concurrentSize));

        linkedQueue.offer(task);
    }

    /**
     * 根据线程名获取对应的待执行的任务
     *
     * @param threadName 线程名称
     * @return 队列内的任务
     */
    private OrderDTO pollTask(String threadName,String bizName)
    {

        int threadNum = Integer.valueOf(threadName.replace(KAFKA_CONSUMER_WORK_THREAD_PREFIX + bizName, ""));

        ConcurrentLinkedQueue<OrderDTO> taskQueue = queues.get(threadNum);

        return taskQueue.poll();
    }
}
